# -*- coding: utf-8 -*-
from datetime import timedelta
from time import time
from utils.operators.rich_sql_sensor import RichSqlSensor
from jms.dm.dm_waybill_collect_intime_plan_summary_dt import jms_dm__dm_waybill_collect_intime_plan_summary_dt

label = f"dm_waybill_collect_intime_plan_summary_dt_{int(time())}"
timeout = timedelta(hours=1).seconds
doris_jms_dm__dm_waybill_collect_intime_plan_summary_dt = RichSqlSensor(
    task_id='doris_jms_dm__dm_waybill_collect_intime_plan_summary_dt',
    pool='broker_load_pool',
    email=['suning@jtexpress.com','yl_bigdata@yl-scm.com'],
    task_concurrency=1,
    conn_id='doris',
    pre_sql=f"""
    TRUNCATE TABLE jms_dm.dm_waybill_collect_intime_plan_summary_dt PARTITION (
        p{{{{ execution_date | cst_ds_nodash }}}}
    );
    LOAD LABEL jms_dm.{label} (
        DATA INFILE(
          "hdfs://{{{{ var.value.hadoop_namespace }}}}/dw/hive/jms_dm.db/external/dm_waybill_collect_intime_plan_summary_dt/dt={{{{ execution_date | cst_ds }}}}/*"
        )   
        INTO TABLE dm_waybill_collect_intime_plan_summary_dt
        FORMAT AS 'PARQUET'(
             network_code
            ,nodal_code
            ,network_agent_code
            ,nodal_agent_code
            ,network_name
            ,network_fran_code
            ,network_fran_name
            ,network_agent_name
            ,nodal_name
            ,nodal_fran_code
            ,nodal_fran_name
            ,nodal_agent_name
            ,jj_cnt
            ,jj_zd_cnt
            ,jj_bzd_cnt
            ,jj_network_cnt
            ,jj_nodal_cnt
            ,effect_qs_cnt
            ,three_code_cnt
            ,error_send_cnt
            ,other_cnt
            ,local_city_cnt
            ,weidao_jj_cnt
            ,route_late_cnt
            ,center_send_timely_cnt
            ,center_send_late_cnt
            ,effect_bzd_cnt
            ,qxt_jj_cnt
            ,qxt_jj_zd_cnt
            ,qxt_jj_bzd_cnt
            ,qxt_jj_network_cnt
            ,qxt_jj_nodal_cnt
            ,qxt_effect_qs_cnt
            ,qxt_three_code_cnt
            ,qxt_error_send_cnt
            ,qxt_other_cnt
            ,qxt_local_city_cnt
            ,qxt_weidao_jj_cnt
            ,qxt_route_late_cnt
            ,qxt_center_send_timely_cnt
            ,qxt_center_send_late_cnt
            ,qxt_effect_bzd_cnt
            ,thd_jj_cnt
            ,thd_jj_zd_cnt
            ,thd_jj_bzd_cnt
            ,thd_jj_network_cnt
            ,thd_jj_nodal_cnt
            ,thd_effect_qs_cnt
            ,thd_three_code_cnt
            ,thd_error_send_cnt
            ,thd_other_cnt
            ,thd_local_city_cnt
            ,thd_weidao_jj_cnt
            ,thd_route_late_cnt
            ,thd_center_send_timely_cnt
            ,thd_center_send_late_cnt
            ,thd_effect_bzd_cnt
            ,zjs_jj_cnt
            ,zjs_jj_zd_cnt
            ,zjs_jj_bzd_cnt
            ,zjs_jj_network_cnt
            ,zjs_jj_nodal_cnt
            ,zjs_effect_qs_cnt
            ,zjs_three_code_cnt
            ,zjs_error_send_cnt
            ,zjs_other_cnt
            ,zjs_local_city_cnt
            ,zjs_weidao_jj_cnt
            ,zjs_route_late_cnt
            ,zjs_center_send_timely_cnt
            ,zjs_center_send_late_cnt
            ,zjs_effect_bzd_cnt
            ,xyf_jj_cnt
            ,xyf_jj_zd_cnt
            ,xyf_jj_bzd_cnt
            ,xyf_jj_network_cnt
            ,xyf_jj_nodal_cnt
            ,xyf_effect_qs_cnt
            ,xyf_three_code_cnt
            ,xyf_error_send_cnt
            ,xyf_other_cnt
            ,xyf_local_city_cnt
            ,xyf_weidao_jj_cnt
            ,xyf_route_late_cnt
            ,xyf_center_send_timely_cnt
            ,xyf_center_send_late_cnt
            ,xyf_effect_bzd_cnt
            ,jdw_jj_cnt
            ,jdw_jj_zd_cnt
            ,jdw_jj_bzd_cnt
            ,jdw_jj_network_cnt
            ,jdw_jj_nodal_cnt
            ,jdw_effect_qs_cnt
            ,jdw_three_code_cnt
            ,jdw_error_send_cnt
            ,jdw_other_cnt
            ,jdw_local_city_cnt
            ,jdw_weidao_jj_cnt
            ,jdw_route_late_cnt
            ,jdw_center_send_timely_cnt
            ,jdw_center_send_late_cnt
            ,jdw_effect_bzd_cnt
            ,qt_jj_cnt
            ,qt_jj_zd_cnt
            ,qt_jj_bzd_cnt
            ,qt_jj_network_cnt
            ,qt_jj_nodal_cnt
            ,qt_effect_qs_cnt
            ,qt_three_code_cnt
            ,qt_error_send_cnt
            ,qt_other_cnt
            ,qt_local_city_cnt
            ,qt_weidao_jj_cnt
            ,qt_route_late_cnt
            ,qt_center_send_timely_cnt
            ,qt_center_send_late_cnt
            ,qt_effect_bzd_cnt
            ,exempt_type
            ,fine_amt
        )
        columns from path as (dt)
        set(
             network_code                       =  network_code
            ,nodal_code                         =  nodal_code
            ,network_agent_code                 =  network_agent_code
            ,nodal_agent_code                   =  nodal_agent_code
            ,network_name                       =  network_name
            ,network_fran_code                  =  network_fran_code
            ,network_fran_name                  =  network_fran_name
            ,network_agent_name                 =  network_agent_name
            ,nodal_name                         =  nodal_name
            ,nodal_fran_code                    =  nodal_fran_code
            ,nodal_fran_name                    =  nodal_fran_name
            ,nodal_agent_name                   =  nodal_agent_name
            ,jj_cnt                             =  jj_cnt
            ,jj_zd_cnt                          =  jj_zd_cnt
            ,jj_bzd_cnt                         =  jj_bzd_cnt
            ,jj_network_cnt                     =  jj_network_cnt
            ,jj_nodal_cnt                       =  jj_nodal_cnt
            ,effect_qs_cnt                      =  effect_qs_cnt
            ,three_code_cnt                     =  three_code_cnt
            ,error_send_cnt                     =  error_send_cnt
            ,other_cnt                          =  other_cnt
            ,local_city_cnt                     =  local_city_cnt
            ,weidao_jj_cnt                      =  weidao_jj_cnt
            ,route_late_cnt                     =  route_late_cnt
            ,center_send_timely_cnt             =  center_send_timely_cnt
            ,center_send_late_cnt               =  center_send_late_cnt
            ,effect_bzd_cnt                     =  effect_bzd_cnt
            ,qxt_jj_cnt                         =  qxt_jj_cnt
            ,qxt_jj_zd_cnt                      =  qxt_jj_zd_cnt
            ,qxt_jj_bzd_cnt                     =  qxt_jj_bzd_cnt
            ,qxt_jj_network_cnt                 =  qxt_jj_network_cnt
            ,qxt_jj_nodal_cnt                   =  qxt_jj_nodal_cnt
            ,qxt_effect_qs_cnt                  =  qxt_effect_qs_cnt
            ,qxt_three_code_cnt                 =  qxt_three_code_cnt
            ,qxt_error_send_cnt                 =  qxt_error_send_cnt
            ,qxt_other_cnt                      =  qxt_other_cnt
            ,qxt_local_city_cnt                 =  qxt_local_city_cnt
            ,qxt_weidao_jj_cnt                  =  qxt_weidao_jj_cnt
            ,qxt_route_late_cnt                 =  qxt_route_late_cnt
            ,qxt_center_send_timely_cnt         =  qxt_center_send_timely_cnt
            ,qxt_center_send_late_cnt           =  qxt_center_send_late_cnt
            ,qxt_effect_bzd_cnt                 =  qxt_effect_bzd_cnt
            ,thd_jj_cnt                         =  thd_jj_cnt
            ,thd_jj_zd_cnt                      =  thd_jj_zd_cnt
            ,thd_jj_bzd_cnt                     =  thd_jj_bzd_cnt
            ,thd_jj_network_cnt                 =  thd_jj_network_cnt
            ,thd_jj_nodal_cnt                   =  thd_jj_nodal_cnt
            ,thd_effect_qs_cnt                  =  thd_effect_qs_cnt
            ,thd_three_code_cnt                 =  thd_three_code_cnt
            ,thd_error_send_cnt                 =  thd_error_send_cnt
            ,thd_other_cnt                      =  thd_other_cnt
            ,thd_local_city_cnt                 =  thd_local_city_cnt
            ,thd_weidao_jj_cnt                  =  thd_weidao_jj_cnt
            ,thd_route_late_cnt                 =  thd_route_late_cnt
            ,thd_center_send_timely_cnt         =  thd_center_send_timely_cnt
            ,thd_center_send_late_cnt           =  thd_center_send_late_cnt
            ,thd_effect_bzd_cnt                 =  thd_effect_bzd_cnt
            ,zjs_jj_cnt                         =  zjs_jj_cnt
            ,zjs_jj_zd_cnt                      =  zjs_jj_zd_cnt
            ,zjs_jj_bzd_cnt                     =  zjs_jj_bzd_cnt
            ,zjs_jj_network_cnt                 =  zjs_jj_network_cnt
            ,zjs_jj_nodal_cnt                   =  zjs_jj_nodal_cnt
            ,zjs_effect_qs_cnt                  =  zjs_effect_qs_cnt
            ,zjs_three_code_cnt                 =  zjs_three_code_cnt
            ,zjs_error_send_cnt                 =  zjs_error_send_cnt
            ,zjs_other_cnt                      =  zjs_other_cnt
            ,zjs_local_city_cnt                 =  zjs_local_city_cnt
            ,zjs_weidao_jj_cnt                  =  zjs_weidao_jj_cnt
            ,zjs_route_late_cnt                 =  zjs_route_late_cnt
            ,zjs_center_send_timely_cnt         =  zjs_center_send_timely_cnt
            ,zjs_center_send_late_cnt           =  zjs_center_send_late_cnt
            ,zjs_effect_bzd_cnt                 =  zjs_effect_bzd_cnt
            ,xyf_jj_cnt                         =  xyf_jj_cnt
            ,xyf_jj_zd_cnt                      =  xyf_jj_zd_cnt
            ,xyf_jj_bzd_cnt                     =  xyf_jj_bzd_cnt
            ,xyf_jj_network_cnt                 =  xyf_jj_network_cnt
            ,xyf_jj_nodal_cnt                   =  xyf_jj_nodal_cnt
            ,xyf_effect_qs_cnt                  =  xyf_effect_qs_cnt
            ,xyf_three_code_cnt                 =  xyf_three_code_cnt
            ,xyf_error_send_cnt                 =  xyf_error_send_cnt
            ,xyf_other_cnt                      =  xyf_other_cnt
            ,xyf_local_city_cnt                 =  xyf_local_city_cnt
            ,xyf_weidao_jj_cnt                  =  xyf_weidao_jj_cnt
            ,xyf_route_late_cnt                 =  xyf_route_late_cnt
            ,xyf_center_send_timely_cnt         =  xyf_center_send_timely_cnt
            ,xyf_center_send_late_cnt           =  xyf_center_send_late_cnt
            ,xyf_effect_bzd_cnt                 =  xyf_effect_bzd_cnt
            ,jdw_jj_cnt                         =  jdw_jj_cnt
            ,jdw_jj_zd_cnt                      =  jdw_jj_zd_cnt
            ,jdw_jj_bzd_cnt                     =  jdw_jj_bzd_cnt
            ,jdw_jj_network_cnt                 =  jdw_jj_network_cnt
            ,jdw_jj_nodal_cnt                   =  jdw_jj_nodal_cnt
            ,jdw_effect_qs_cnt                  =  jdw_effect_qs_cnt
            ,jdw_three_code_cnt                 =  jdw_three_code_cnt
            ,jdw_error_send_cnt                 =  jdw_error_send_cnt
            ,jdw_other_cnt                      =  jdw_other_cnt
            ,jdw_local_city_cnt                 =  jdw_local_city_cnt
            ,jdw_weidao_jj_cnt                  =  jdw_weidao_jj_cnt
            ,jdw_route_late_cnt                 =  jdw_route_late_cnt
            ,jdw_center_send_timely_cnt         =  jdw_center_send_timely_cnt
            ,jdw_center_send_late_cnt           =  jdw_center_send_late_cnt
            ,jdw_effect_bzd_cnt                 =  jdw_effect_bzd_cnt
            ,qt_jj_cnt                          =  qt_jj_cnt
            ,qt_jj_zd_cnt                       =  qt_jj_zd_cnt
            ,qt_jj_bzd_cnt                      =  qt_jj_bzd_cnt
            ,qt_jj_network_cnt                  =  qt_jj_network_cnt
            ,qt_jj_nodal_cnt                    =  qt_jj_nodal_cnt
            ,qt_effect_qs_cnt                   =  qt_effect_qs_cnt
            ,qt_three_code_cnt                  =  qt_three_code_cnt
            ,qt_error_send_cnt                  =  qt_error_send_cnt
            ,qt_other_cnt                       =  qt_other_cnt
            ,qt_local_city_cnt                  =  qt_local_city_cnt
            ,qt_weidao_jj_cnt                   =  qt_weidao_jj_cnt
            ,qt_route_late_cnt                  =  qt_route_late_cnt
            ,qt_center_send_timely_cnt          =  qt_center_send_timely_cnt
            ,qt_center_send_late_cnt            =  qt_center_send_late_cnt
            ,qt_effect_bzd_cnt                  =  qt_effect_bzd_cnt
            ,exempt_type                        =  exempt_type
            ,fine_amt                           =  fine_amt
            ,dt                                 =  dt
        )
    )
    WITH BROKER '{{{{ var.json.doris_brokers | random_choice }}}}'
    PROPERTIES ('timeout'='{timeout}', 'max_filter_ratio'='0.0')""",
    poke_sql=f"SHOW LOAD FROM jms_dm WHERE label = '{label}' ORDER BY CreateTime DESC LIMIT 1",
    sql_on_kill=f"CANCEL LOAD FROM jms_dm WHERE LABEL = '{label}'",
    success=lambda r: r[2] == 'FINISHED',
    failure=lambda r: (r[2] is not None and r[2] == 'CANCELLED', str(r[7])),
    poke_interval=60,
    execution_timeout=timedelta(seconds=timeout + 120), )

doris_jms_dm__dm_waybill_collect_intime_plan_summary_dt << jms_dm__dm_waybill_collect_intime_plan_summary_dt
